package com.jacquesh;

import com.jacquesh.model.CoffeeShop;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.util.Collector;

import java.text.SimpleDateFormat;

public class FlinkDemo0005 {
    public static void main(String[] args) throws Exception {
        SimpleDateFormat format = new SimpleDateFormat("yyyy-MM-dd");
        StreamExecutionEnvironment environment =
                StreamExecutionEnvironment.getExecutionEnvironment();
        DataStreamSource<String> stream = environment
                .socketTextStream("192.168.88.180", 8888);
        DataStream dataStream = stream.
                flatMap(new RichFlatMapFunction<String, CoffeeShop>() {
            @Override
            public void flatMap(String s, Collector<CoffeeShop> collector) {
                try {
                    String[] sp = s.split(",");
                    collector.collect(CoffeeShop.builder()
                            .id(Long.parseLong(sp[0]))
                            .date(format.parse(sp[1]))
                            .sales(Integer.parseInt(sp[2]))
                            .build());
                } catch (Exception ex) {
                    System.out.println(ex);
                }

            }
        })        /** 依据id分组*/
                .keyBy(CoffeeShop::getId).reduce(new ReduceFunction<CoffeeShop>() {
                    @Override
                    public CoffeeShop reduce(CoffeeShop coffeeShop, CoffeeShop t1) throws Exception {
                        /***
                         * 聚合 分区后 只保留销量大
                         */
                        return new CoffeeShop(coffeeShop.getId(),
                                coffeeShop.getDate(),
                                Math.max(coffeeShop.getSales(),
                                t1.getSales())
                        );
                    }
                });
        dataStream.print().setParallelism(1);
        environment.execute("reduce demo");
    }
}
